<?php
/**
 * OMadapter - PHP Web Server, CRM adapter
 *
 * Receive message from OM server and dispatch to crm
 * Send control message to OM
 *
 */
 namespace OM;

 require_once dirname( __FILE__ ) . '/Lib/Constants.php';

 use OM\Events\EventInterface;
 use OM\Lib\Connection\ConnectionInterface;
 use OM\Lib\Protocols\Http;
 use OM\Lib\Protocols\HttpCache;
 use OM\Lib\Protocols\Websocket;
 use OM\Lib\Connection\TcpConnection;
 use OM\Dev\Om;
 use OM\Dev\Ext;
 use OM\Lib\Timer;


use Exception;

 class OMadapter
 {
 	/**
 	 * version
 	 */
 	const VERSION = '0.1';

 	const STATUS_STARTING = 1;
    const STATUS_RUNNING = 2;
    const STATUS_SHUTDOWN = 4;
    const STATUS_RELOADING = 8;

    const LOG_ERR = "ERR \t";
    const LOG_NOTICE = "NOTICE \t";
    const LOG_WARN = "WARNING\t";
    const LOG_DEBUG = "DEBUG \t";
    const LOG_MSG = "MSG  \t";
    const KILL_WORKER_TIMER_TIME = 2;
    /**
     * Default backlog. Backlog is the maximum length of the queue of pending connections.
     *
     * @var int
     */
    const DEFAUL_BACKLOG = 1024;

    //variable
    public $m_id ;
    public $m_processname = 'om';
    public $m_count = 1 ;
    public $m_user = '';
    public $m_group = '';
    public $m_name = "OM adapter";

    public $onStart = null;

    public $onConnect = null;
    public $onMessage = null;
    public $onClose = null;
    public $onError = null;
    public $onBufferFull = null;
    public $onBufferDrain = null;
    public $onTest = null;

    //websocket 事件绑定
    public $onWebSocketConnect = null;

    public $onStop = null;
    public $onReload = null;

    /*所有的tcp链接*/
    public $m_connections = array();
    /* 所有的websocket 链接 */
    public $m_wsConnections = array();
    public $reusePort = false; //socket port reuse
    protected $_autoloadRootPath = '';

    public static $daemonize = false;
    public static $stdoutFile = '/dev/null';
    public static $m_runtime_dir = "./runtime/";
    public static $logFile = '';
    public static $pidFile = '';

    protected static $defaultMimeType = 'text/html; charset=utf-8';
    protected static $mimeTypeMap = array();
    public static $om_notify_url = '/notify'; //OM device send notify message address
    public static $om_ws_url = '/ws'; //websocket连接地址
    public static $om_test_url = '/test'; //测试 URL
    public static $OM_CRM_URL = CRM_URL;//CRM服务器地址

    public static $globalEvent = null;
    protected static $_masterPid = 0;
    protected $_mainSocket = null;
    protected $_socketName = '';
    protected $_context = null;

    protected static $_workers = array();
    protected static $_pidMap = array();
    protected static $_pidsToRestart = array();
    protected static $_idMap = array();

    /**
     * OM设备信息
     */
    public static $om_devices = array();
    public static $om_ipMap = array();


    protected static $_status = self::STATUS_STARTING;

    protected static $_maxWorkerNameLength = 12;

    /**
     * Maximum length of the socket names.
     */
    protected static $_maxSocketNameLength = 12;

    /**
     * Maximum length of the process user names.
     */
    protected static $_maxUserNameLength = 12;


  	protected static $_statisticsFile = '';
    protected static $_globalStatistics = array(
        'start_timestamp'  => 0,
        'worker_exit_info' => array()
    );
    protected static $_startFile = '';


 	//run OM server
 	public static function runAll()
 	{
 		self::init();
 		self::parseCmd();
 		self::daemonize();
 		self::initWorker();
 		self::installSignal();
        self::saveMasterPid();
        self::runWorker();
 		self::showUI();
 		self::resetStd();
 		self::monitor();
 	}

    //初始化函数
    public function __construct($socket_name = '', $context_option = array())
    {
        // Save all worker instances.
        $this->m_id                 = spl_object_hash($this);
        self::$_workers[$this->m_id] = $this;
        self::$_pidMap[$this->m_id]  = array();

        // Get autoload root path.
        $backrace                = debug_backtrace();
        $this->_autoloadRootPath = dirname($backrace[0]['file']);

        // Context for socket.
        if ($socket_name) {
            $this->_socketName = $socket_name;
            if (!isset($context_option['socket']['backlog'])) {
                $context_option['socket']['backlog'] = self::DEFAUL_BACKLOG;
            }
            $this->_context = stream_context_create($context_option);
        }

        //tcp断开链接的时候
        $this->onClose = function( $connection )
        {
            $tcpID = $connection->id;
            if( isset( $this->m_wsConnections[$tcpID] ) ){
                $omID = $this->m_wsConnections[$tcpID]['omID'];
                $subID = $this->m_wsConnections[$tcpID]['omID'];
                unset($this->m_wsConnections[$tcpID] );
                self::$om_devices[$omID]->wsOffLineExt( $subID );
            }
        };

        //websocket客户端链接上来的时候，服务器发送初始化数据
        $this->onWebSocketConnect = function( $connection )
        {
            if( !$this->m_wsConnections[$connection->id]['isShake'] )
                $this->m_wsConnections[$connection->id]['isShake']  = true;
            $extData = $this->m_wsConnections[$connection->id];

            //给客户端发送初始化信息
            $omID = $extData&&isset( $extData['omID'] )?  $extData['omID'] : false;
            if( $omID ) {
                $res = self::$om_devices[$omID]->InitExt2Ws( $extData['subID'] ,null );
                if( $res ){
                    self::$om_devices[$omID]->m_exts[ $extData['subID']]->ws_state = 1;
                    $connection->send( json_encode( $res ));
                }
                else{
                    $res = array('rcode'=>-1,'data'=>NULL,'msg'=>"初始化分机信息失败");
                    $connection->close( json_encode( $res ));
                }
            }
            else
                $connection->close("error");
            return;
        };

        //当有外部消息推送过来的时候
        $this->onMessage = function ( $connection , $sock_data )
        {
            //websocket消息
            if( isset( $_SERVER['IS_WS']) ){
                $extData = $this->m_wsConnections[$connection->id];
                $extID = $extData?  $extData['subID'] : false;
                $omID =  $extData?  $extData['omID'] : false;
                if( !$extID || !$omID || !isset( self::$om_devices[$omID] ) )
                {
                    $errData = array('rcode'=>0,'data'=>null,'msg'=>'parse data error');
                    $connection->send( json_encode( $errData ) );
                    return;
                }
                self::$om_devices[$omID]->notifyWebSocket( $extID, $sock_data , $connection );
                if( !$this->m_wsConnections[$connection->id]['isShake']  )
                    $this->m_wsConnections[$connection->id]['isShake'] = true;
                return;
            }
            //根据请求的URL地址，转发请求
            $OM_url_info = parse_url($_SERVER['REQUEST_URI']);
            if (!$OM_url_info) {
                Http::header('HTTP/1.1 400 Bad Request');
                $connection->close('<h1>400 Bad Request</h1>');
                return;
            }

            $OM_path = isset($OM_url_info['path']) ? $OM_url_info['path'] : '/';
            //首页访问测试地址
            if( '/' === $OM_path || '' === $OM_path ){
                Http::header('HTTP/1.1 200 OK');
                $connection->close("OK");
                return ;
            }
            elseif( self::$om_notify_url === $OM_path ){
                //根据推送服务器的地址判断OM编号，获取OM对象
                $om_addr = $_SERVER['REMOTE_ADDR'] ? $_SERVER['REMOTE_ADDR'] : false;
                if( $om_addr ){
                    $ipMap = array_flip( self::$om_ipMap );
                    $omID = isset( $ipMap[$om_addr]) ?  $ipMap[$om_addr] : false;
                    if( self::$om_devices[$omID]->notifyParse(  $GLOBALS['HTTP_RAW_POST_DATA'] ) ){
                        Http::header('HTTP/1.1 200 OK');
                        $connection->close("done");
                        return;
                    }
                }
                Http::header('HTTP/1.1 500 Error');
                $connection->close("err");
                return ;
            }
            elseif( self::$om_ws_url === $OM_path )
            {
                $connID = $connection->id;
                /**
                 * 本地没有保存webcoket链接首先握手,websocket链接首次一定是进入这里
                 * 可以参考websocket客户端与服务器握手流程
                 */
                if( !isset( $this->m_wsConnections[$connID] ) )
                {
                    $doHandshake = false;
                    $subID = isset( $_GET['subID'] ) ? trim( $_GET['subID'] ) : false;
                    $omID = isset( $_GET['omID'] ) ? trim( $_GET['omID'] ) : false;
                    $crm_uid = isset( $_GET['UID'] ) ? trim( $_GET['UID'] ) : false; //用户登陆CRM之后的用户ID
                    if( $subID && $omID && isset( self::$om_devices[$omID] ) && isset( self::$om_devices[$omID]->m_exts[$subID] ) )
                    {
                        $doHandshake = true;
                        //该账号已经在别的地方登录,直接把以前的踢掉
                        /*if( self::$om_devices[$omID]->m_exts[$subID]->ws_state ){
                            $doHandshake = true;
                            foreach( $this->m_wsConnections as $kc=>$vc){
                                if( $subID == $vc['subID'] ){
                                    $extOffMsg = array('op'=> Ext::API_OFFLINE, 'msg'=> Ext::ERR_MSG_RECONNECT );
                                    $this->m_connections[$kc]->close( json_encode( $extOffMsg ));
                                    self::$om_devices[$omID]->m_exts[$subID]->ws_state  = 0;
                                    break;
                                }
                            }
                        }*/
                        self::$om_devices[$omID]->m_exts[$subID]->crm_uid  = $crm_uid;
                    }
                    else{
                        self::log( self::LOG_ERR, 'SubID:'.$subID.' OMID:'.$omID.'  crm_uid:'.$crm_uid.' ws connect failed');
                    }
                    if( $doHandshake){
                        //根据请求的 URL参数设置分机号、OM设备编号
                        $this->m_wsConnections[$connID] = array('isShake' => false,'subID'=>$subID,'omID'=>$omID, 'uid'=>$crm_uid);
                        self::$om_devices[$omID]->m_exts[$subID]->ws_state  = 1;//设置分机的websocket登陆状态
                        $this->m_connections[$connID]->protocol = "\\OM\Lib\\Protocols\\Websocket";
                        Websocket::dealHandshake(  $_SERVER["SW_KEY"], $connection );
                    }
                    else{
                        self::log( self::LOG_ERR, 'SubID:'.$subID.' OMID:'.$omID.'  crm_uid:'.$crm_uid.' Websocket握手失败 ');
                        Http::header('HTTP/1.1 400 Bad Request');
                        $connection->close( Ext::ERR_MSG_INIT );
                    }
                }
                else{
                    self::log( self::LOG_ERR, 'SubID:'.$subID.' OMID:'.$omID.'  crm_uid:'.$crm_uid.' 获取链接信息失败');
                    Http::header('HTTP/1.1 400 Bad Request');
                    $connection->close(  Ext::ERR_MSG_INIT );
                }
                return ;
            }
            //
            elseif( OMadapter::$om_test_url === $OM_path )
            {
                if( $this->onTest && OM_DEBUG )
                {
                    try {
                            call_user_func($this->onTest, $connection );
                        } catch (\Exception $e) {
                            self::log($e);
                            exit(250);
                        } catch (\Error $e) {
                            self::log($e);
                            exit(250);
                        }
                }

                if( strtolower($_SERVER['REQUEST_METHOD']) == 'get' )
                {
                    Http::header('HTTP/1.1 200 GET OK');
                    $connection->close('');
                }
                else
                {
                    Http::header('HTTP/1.1 200 POST OK');
                    $connection->close('');
                }
                return ;

            }
            else
            {
                if( !OM_DEBUG )
                {
                    Http::header('HTTP/1.1 400 Bad Request');
                    $connection->close();
                    return;
                }
                $OM_path_info      = pathinfo($OM_path);
                $om_file_extension = isset($OM_path_info['extension']) ? $OM_path_info['extension'] : '';

                $worker_root_dir  = dirname( self::$_startFile ).'/';
                $om_worker_file = "$worker_root_dir/$OM_path";

                // File exsits.
                if (is_file($om_worker_file))
                {
                    // Security checkom_worker_file
                    if ((!($workerman_request_realpath = realpath($om_worker_file)) || !($workerman_root_dir_realpath = realpath($worker_root_dir))) || 0 !== strpos($workerman_request_realpath,
                            $workerman_root_dir_realpath)
                        )
                    {
                        Http::header('HTTP/1.1 400 Bad Request');
                        $connection->close('<h1>400 Bad Request</h1>');
                        return;
                    }

                    // Request php file.
                    if ($om_file_extension === 'php')
                    {
                        $workerman_cwd = getcwd();
                        chdir($worker_root_dir);
                        ini_set('display_errors', 'off');
                        ob_start();
                        // Try to include php file.
                        try {
                            // $_SERVER.
                            $_SERVER['REMOTE_ADDR'] = $connection->getRemoteIp();
                            $_SERVER['REMOTE_PORT'] = $connection->getRemotePort();
                            include $om_worker_file;
                        } catch (\Exception $e) {
                            // Jump_exit?
                            if ($e->getMessage() != 'jump_exit') {
                                echo $e;
                            }
                        }
                        $content = ob_get_clean();
                        ini_set('display_errors', 'on');
                        $connection->close($content);
                        chdir($workerman_cwd);
                        return;
                    }

                    // Send file to client.
                    return self::sendFile($connection, $om_worker_file);
                }
                else
                {
                    // 404
                    Http::header("HTTP/1.1 404 Not Found");
                    $connection->close('<html><head><title>404 File not found</title></head><body><center><h3>404 Not Found</h3></center></body></html>');
                    return;
                }

                Http::header('HTTP/1.1 403 Forbidden');
                $connection->close('error');
            }
            return;
        };
    }

    public static function sendFile($connection, $file_name)
    {
        // Check 304.
        $info = stat($file_name);
        $modified_time = $info ? date('D, d M Y H:i:s', $info['mtime']) . ' GMT' : '';
        if (!empty($_SERVER['HTTP_IF_MODIFIED_SINCE']) && $info) {
            // Http 304.
            if ($modified_time === $_SERVER['HTTP_IF_MODIFIED_SINCE']) {
                // 304
                Http::header('HTTP/1.1 304 Not Modified');
                // Send nothing but http headers..
                $connection->close('');
                return;
            }
        }

        // Http header.
        if ($modified_time) {
            $modified_time = "Last-Modified: $modified_time\r\n";
        }
        $file_size = filesize($file_name);
        $extension = pathinfo($file_name, PATHINFO_EXTENSION);
        $content_type = isset(self::$mimeTypeMap[$extension]) ? self::$mimeTypeMap[$extension] : self::$defaultMimeType;
        $header = "HTTP/1.1 200 OK\r\n";
        $header .= "Content-Type: $content_type\r\n";
        $header .= "Connection: keep-alive\r\n";
        $header .= $modified_time;
        $header .= "Content-Length: $file_size\r\n\r\n";
        $trunk_limit_size = 1024*1024;
        if ($file_size < $trunk_limit_size) {
            return $connection->send($header.file_get_contents($file_name), true);
        }
        $connection->send($header, true);

        // Read file content from disk piece by piece and send to client.
        $connection->fileHandler = fopen($file_name, 'r');
        $do_write = function()use($connection)
        {
            // Send buffer not full.
            while(empty($connection->bufferFull))
            {
                // Read from disk.
                $buffer = fread($connection->fileHandler, 8192);
                // Read eof.
                if($buffer === '' || $buffer === false)
                {
                    return;
                }
                $connection->send($buffer, true);
            }
        };
        // Send buffer full.
        $connection->onBufferFull = function($connection)
        {
            $connection->bufferFull = true;
        };
        // Send buffer drain.
        $connection->onBufferDrain = function($connection)use($do_write)
        {
            $connection->bufferFull = false;
            $do_write();
        };
        $do_write();
    }

     public static function initMimeTypeMap()
    {
        $mime_file = Http::getMimeTypesFile();
        if (!is_file($mime_file)) {
            self::log("$mime_file mime.type file not fond");
            return;
        }
        $items = file($mime_file, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES);
        if (!is_array($items)) {
            self::log("get $mime_file mime.type content fail");
            return;
        }
        foreach ($items as $content) {
            if (preg_match("/\s*(\S+)\s+(\S.+)/", $content, $match)) {
                $mime_type                      = $match[1];
                $workerman_file_extension_var   = $match[2];
                $workerman_file_extension_array = explode(' ', substr($workerman_file_extension_var, 0, -1));
                foreach ($workerman_file_extension_array as $workerman_file_extension) {
                    self::$mimeTypeMap[$workerman_file_extension] = $mime_type;
                }
            }
        }
    }

 	//init
 	public static function init()
 	{
 		self::checkEnv();
 		$backtrace        = debug_backtrace();
        self::$_startFile = $backtrace[count($backtrace) - 1]['file'];

        $runtimeDir = str_replace("//","/",dirname( self::$_startFile).'/'.self::$m_runtime_dir .'/'  );
        if( !file_exists( $runtimeDir ))
        {
            mkdir( $runtimeDir, 0755, true);
        }
        self::$m_runtime_dir = $runtimeDir;

        // Pid file.
        if (empty(self::$pidFile)) {
            self::$pidFile = self::$m_runtime_dir. str_replace('/', '_', dirname(self::$_startFile) ) . ".pid";
        }

        // Log file.
        if (empty(self::$logFile)) {
            self::$logFile = self::$m_runtime_dir. 'OM'.date('Y_m_d').'.log';
        }

        touch(self::$logFile);
        chmod(self::$logFile, 0622);

        self::initMimeTypeMap();
        // State.
        self::$_status = self::STATUS_STARTING;
        self::$_globalStatistics['start_timestamp'] = time();
        self::$_statisticsFile                      = self::$m_runtime_dir. 'OM.status';

        // Process title.
        self::setProcessTitle('OM: master process  start_file=' . self::$_startFile);

        self::initId();

        Timer::init();
 	}

 	//init  workder
 	public static function initWorker()
 	{
        foreach (self::$_workers as $worker) {
            // Worker name.
            if (empty($worker->m_name)) {
                $worker->m_name = 'none';
            }

            // Get maximum length of worker name.
            $worker_name_length = strlen($worker->m_name);
            if (self::$_maxWorkerNameLength < $worker_name_length) {
                self::$_maxWorkerNameLength = $worker_name_length;
            }

            // Get maximum length of socket name.
            $socket_name_length = strlen($worker->getSocketName());
            if (self::$_maxSocketNameLength < $socket_name_length) {
                self::$_maxSocketNameLength = $socket_name_length;
            }

            // Get unix user of the worker process.
            if (empty($worker->user)) {
                $worker->user = self::getCurrentUser();
            } else {
                if (posix_getuid() !== 0 && $worker->user != self::getCurrentUser()) {
                    self::log('Warning: You must have the root privileges to change uid and gid.');
                }
            }

            // Get maximum length of unix user name.
            $user_name_length = strlen($worker->user);
            if (self::$_maxUserNameLength < $user_name_length) {
                self::$_maxUserNameLength = $user_name_length;
            }

            // Listen.
            if (!$worker->reusePort) {
                $worker->listen();
            }
        }
 	}

 	//parse cli command parameter
 	public static function parseCmd()
 	{
 		 global $argv;
        // Check argv;
        $start_file = $argv[0];
        if (!isset($argv[1])) {
            exit("Usage: php yourfile.php {start|stop|restart|reload|status|kill}\n");
        }

        // Get command.
        $command  = trim($argv[1]);
        $command2 = isset($argv[2]) ? $argv[2] : '';

        // Start command.
        $mode = '';
        if ($command === 'start') {
            if ($command2 === '-d') {
                $mode = 'in DAEMON mode';
            } else {
                $mode = 'in DEBUG mode';
            }
        }
        self::log(self::LOG_NOTICE."OM[$start_file] $command $mode");

        // Get master process PID.
        $master_pid      = @file_get_contents(self::$pidFile);
        $master_is_alive = $master_pid && @posix_kill($master_pid, 0);
        // Master is still alive?
        if ($master_is_alive) {
            if ($command === 'start') {
                self::log(self::LOG_WARN."OM[$start_file] already running");
                exit;
            }
        } elseif ($command !== 'start' && $command !== 'restart' && $command !== 'kill') {
            self::log(self::LOG_NOTICE."OM[$start_file] not run");
            exit;
        }

        // execute command.
        switch ($command) {
            case 'kill':
                exec("ps aux | grep $start_file | grep -v grep | awk '{print $2}' |xargs kill -SIGINT");
                usleep(100000);
                exec("ps aux | grep $start_file | grep -v grep | awk '{print $2}' |xargs kill -SIGKILL");
                break;
            case 'start':
                if ($command2 === '-d') {
                    OMadapter::$daemonize = true;
                }
                break;
            case 'status':
                if (is_file(self::$_statisticsFile)) {
                    @unlink(self::$_statisticsFile);
                }
                // Master process will send status signal to all child processes.
                posix_kill($master_pid, SIGUSR2);
                // Waiting amoment.
                usleep(100000);
                // Display statisitcs data from a disk file.
                @readfile(self::$_statisticsFile);
                exit(0);
            case 'restart':
            case 'stop':
                self::log("OM[$start_file] is stoping ...");
                // Send stop signal to master process.
                $master_pid && posix_kill($master_pid, SIGINT);
                // Timeout.
                $timeout    = 5;
                $start_time = time();
                // Check master process is still alive?
                while (1) {
                    $master_is_alive = $master_pid && posix_kill($master_pid, 0);
                    if ($master_is_alive) {
                        // Timeout?
                        if (time() - $start_time >= $timeout) {
                            self::log("OM[$start_file] stop fail");
                            exit;
                        }
                        // Waiting amoment.
                        usleep(10000);
                        continue;
                    }
                    // Stop success.
                    self::log("OM[$start_file] stop success");
                    if ($command === 'stop') {
                        exit(0);
                    }
                    if ($command2 === '-d') {
                        OMadapter::$daemonize = true;
                    }
                    break;
                }
                break;
            case 'reload':
                posix_kill($master_pid, SIGUSR1);
                self::log("OM[$start_file] reload");
                exit;
            default :
                exit("Usage: php yourfile.php {start|stop|restart|reload|status|kill}\n");
        }
 	}

 	//display comand
 	public static function showUI()
 	{
 		echo "\033[1A\n\033[K-----------------------\033[47;30m OMadapter \033[0m-----------------------------\n\033[0m";
        echo 'version:', OMadapter::VERSION, "          PHP version:", PHP_VERSION, "        Memory:",ini_get('memory_limit'),"\n";
        echo "------------------------\033[47;30m WORKERS \033[0m-------------------------------\n";
        echo "\033[47;30muser\033[0m", str_pad('',
            self::$_maxUserNameLength + 2 - strlen('user')), "\033[47;30mworker\033[0m", str_pad('',
            self::$_maxWorkerNameLength + 2 - strlen('worker')), "\033[47;30mlisten\033[0m", str_pad('',
            self::$_maxSocketNameLength + 2 - strlen('listen')), "\033[47;30mprocesses\033[0m \033[47;30m", "status\033[0m\n";

        foreach (self::$_workers as $worker) {
            echo str_pad($worker->user, self::$_maxUserNameLength + 2), str_pad($worker->m_name,
                self::$_maxWorkerNameLength + 2), str_pad($worker->getSocketName(),
                self::$_maxSocketNameLength + 2), str_pad(' ' . $worker->m_count, 9), " \033[32;40m [OK] \033[0m\n";;
        }
        echo "------------------------\033[47;30m OMS \033[0m-------------------------------\n";
        if( count(self::$om_ipMap ) ){
            foreach( self::$om_ipMap as $kom=>$vom ){
                if( self::$om_devices[$kom]->states == Om::OM_RUNNING){
                    echo "OM ID:".str_pad($kom,5)."\t addr:".str_pad($vom,16) . str_pad("\033[32;40m RUNNIG\033[0m", 20). "\n";
                }else{
                    echo "OM ID:".str_pad($kom,5)."\t addr:".str_pad($vom,16) . str_pad("\033[31;40m OFFLINE\033[0m", 20). "\n";
                }
            }
        }
        else
            echo "No OM Devices\n";
        echo "----------------------------------------------------------------\n";
        if (self::$daemonize) {
            global $argv;
            $start_file = $argv[0];
            echo "Input \"php $start_file stop\" to quit. Start success.\n";
        } else {
            echo "Press Ctrl-C to quit. Start success.\n";
        }
 	}

 	//run all worker
 	public static function runWorker()
 	{
        //初始化OM配置信息
        self::initOMs();
 		foreach (self::$_workers as $worker) {
            if (self::$_status === self::STATUS_STARTING) {
                if (empty($worker->m_name)) {
                    $worker->name = $worker->getSocketName();
                }
                $worker_name_length = strlen($worker->m_name);
                if (self::$_maxWorkerNameLength < $worker_name_length) {
                    self::$_maxWorkerNameLength = $worker_name_length;
                }
            }

            while (count(self::$_pidMap[$worker->m_id]) < $worker->m_count) {
                static::forkOneWorker($worker);
            }
        }
 	}

 	 /**
     * Fork one worker process.
     *
     * @param Worker $worker
     * @throws Exception
     */
    protected static function forkOneWorker($worker)
    {
        $pid = pcntl_fork();
        // Get available worker id.
        $id = self::getId($worker->m_id, 0);
        // For master process.
        if ($pid > 0) {
            self::$_pidMap[$worker->m_id][$pid] = $pid;
            self::$_idMap[$worker->m_id][$id]   = $pid;
        } // For child processes.
        elseif (0 === $pid) {
            if ($worker->reusePort) {
                $worker->listen();
            }
            if (self::$_status === self::STATUS_STARTING) {
                self::resetStd();
            }
            self::$_pidMap  = array();
            self::$_workers = array($worker->m_id => $worker);
            Timer::delAll();
            self::setProcessTitle('OM: worker process  ' . $worker->m_name . ' ' . $worker->getSocketName());
            $worker->setUserAndGroup();
            $worker->id = $id;
            $worker->run();
            exit(250);
        } else {
            throw new Exception("forkOneWorker fail");
        }
    }

    public function setUserAndGroup()
    {
        return;
        // Get uid.
        $user_info = posix_getpwnam($this->m_user);
        if (!$user_info) {
            self::log("Warning: User {$this->user} not exsits");
            return;
        }
        $uid = $user_info['uid'];
        // Get gid.
        if ($this->m_group) {
            $group_info = posix_getgrnam($this->m_group);
            if (!$group_info) {
                self::log("Warning: Group {$this->group} not exsits");
                return;
            }
            $gid = $group_info['gid'];
        } else {
            $gid = $user_info['gid'];
        }

        // Set uid and gid.
        if ($uid != posix_getuid() || $gid != posix_getgid()) {
            if (!posix_setgid($gid) || !posix_initgroups($user_info['name'], $gid) || !posix_setuid($uid)) {
                self::log("Warning: change gid or uid fail.");
            }
        }
    }

 	//monitor
 	public static function monitor()
 	{
 		self::$_status = self::STATUS_RUNNING;
        while (1) {
            // Calls signal handlers for pending signals.
            pcntl_signal_dispatch();
            // Suspends execution of the current process until a child has exited, or until a signal is delivered
            $status = 0;
            $pid    = pcntl_wait($status, WUNTRACED);
            // Calls signal handlers for pending signals again.
            pcntl_signal_dispatch();
            // If a child has already exited.
            if ($pid > 0) {
                // Find out witch worker process exited.
                foreach (self::$_pidMap as $worker_id => $worker_pid_array) {
                    if (isset($worker_pid_array[$pid])) {
                        $worker = self::$_workers[$worker_id];
                        // Exit status.
                        if ($status !== 0) {
                            self::log("worker[" . $worker->m_name . ":$pid] exit with status $status");
                        }

                        // For Statistics.
                        if (!isset(self::$_globalStatistics['worker_exit_info'][$worker_id][$status])) {
                            self::$_globalStatistics['worker_exit_info'][$worker_id][$status] = 0;
                        }
                        self::$_globalStatistics['worker_exit_info'][$worker_id][$status]++;

                        // Clear process data.
                        unset(self::$_pidMap[$worker_id][$pid]);

                        // Mark id is available.
                        $id                            = self::getId($worker_id, $pid);
                        self::$_idMap[$worker_id][$id] = 0;

                        break;
                    }
                }
                // Is still running state then fork a new worker process.
                if (self::$_status !== self::STATUS_SHUTDOWN) {
                    self::forkWorkers();
                    // If reloading continue.
                    if (isset(self::$_pidsToRestart[$pid])) {
                        unset(self::$_pidsToRestart[$pid]);
                        self::reload();
                    }
                } else {
                    // If shutdown state and all child processes exited then master process exit.
                    if (!self::getAllWorkerPids()) {
                        self::exitAndClearAll();
                    }
                }
            } else {
                // If shutdown state and all child processes exited then master process exit.
                if (self::$_status === self::STATUS_SHUTDOWN && !self::getAllWorkerPids()) {
                    self::exitAndClearAll();
                }
            }
        }
 	}

    //从配置文件中加载OM信息
    protected static function initOMs()
    {
        /*从CRM服务器加载设备信息*/
        $_arrOMs = Om::getOmList( true );
        if( !$_arrOMs ){
            self::log(self::LOG_ERR."从远程CRM加载配置信息失败\n");
            exit("从远程数据库加载OM配置信息失败，请检查CRM中OM配置信息\n");
        }

        foreach( $_arrOMs as $vv ){//初始化配置信息
            if( !isset( self::$om_devices[$vv['omid']] ) ){
                $_OM = new Om($vv['omid'], $vv['serverip'], $vv['serviceport']);
                $_OM->ipaddr = $vv['ipaddr'];
                $_OM->mp3_server = $vv['mp3_server'];
                $_OM->mp3_dir = $vv['mp3_dir'];
                $_OM->dbhost = $vv['dbhost'];
                $_OM->dbport = $vv['dbport'] ? $vv['dbport'] : 3306;
                $_OM->dbname = $vv['dbname'];
                $_OM->dbuser = $vv['dbuser'];
                $_OM->dbpsw = $vv['dbpsw'];
                self::$om_devices[$vv['omid']] = $_OM;
                self::$om_ipMap[$vv['omid']] = $vv['serverip'];
            }
        }
    }

 	protected static function initId()
    {
        foreach (self::$_workers as $worker_id => $worker) {
            self::$_idMap[$worker_id] = array_fill(0, $worker->m_count, 0);
        }
    }

    protected static function getId($worker_id, $pid)
    {
        $id = array_search($pid, self::$_idMap[$worker_id]);
        if ($id === false) {
            echo "getId fail\n";
        }
        return $id;
    }

 	//适配器运行环境检查
 	protected static function  checkEnv()
 	{
 		//cli运行
		if (php_sapi_name() != "cli") {
		    exit("only run in command line mode \n");
		}

        //检查是否支持event 事件模型
        if( !extension_loaded( "event" ) )
        {
            $helpMsg = "Please ref:https://bitbucket.org/osmanov/pecl-event/src/e2369dac708c910c43b7b7e55a4219ddc3f53896/INSTALL.md?fileviewer=file-view-default \n";
            $helpMsg .= 'Extension Download URL:https://pecl.php.net/package/event';

            exit("OM need php event extension,please enable or install php event extension!\n".$helpMsg );
        }

        //检查是佛支持curl扩展，用户向OM推送控制命令
        if( !extension_loaded( "curl" ) )
        {
            exit("curl need php curl extension");
        }
 	}

    /**
     * 向指定分机发送消息
     *
     * @param  string $_omID  OM适配器编号
     * @param  string $_extID 分机号
     * @param  string $_data  向客户端最终发送的数据
     * @return boolean        是否发送成功
     */
    public static function sendWs2Ext( $_omID, $_extID, $_data ){
        if( !self::$_workers || !is_array( self::$_workers ) )
            return false;
        foreach( self::$_workers as $vworker ){
            foreach( $vworker->m_wsConnections  as $conKey=>$conVal ){
                if( $conVal['omID'] == $_omID && $conVal['subID'] == $_extID){
                    if( isset($vworker->m_connections[$conKey]) && $conVal['isShake']  ){
                        $vworker->m_connections[$conKey]->send( $_data);
                        return true;
                    }
                    break;
                }
            }
        }
        return false;
    }
    /**
     * 向已经连接的websocket 客户端发送消息
     *
     * @param  string $_omID   OM适配器编号
     * @param  string   $_data   通过websocket发送的json数据
     * @return boolean/int     false,发生错误;int 已经发送消息的对象
     */
    public static function sendWsBroadCast($_omID, $_data )
    {
        if( !self::$_workers || !is_array( self::$_workers ) )
            return false;
        $sendCount = 0;
        foreach( self::$_workers as $vworker ){
            foreach( $vworker->m_wsConnections  as $conKey=>$conVal ){
                if( $conVal['omID'] == $_omID && $conVal['isShake'] ){
                    if( isset($vworker->m_connections[$conKey]) ){
                        $vworker->m_connections[$conKey]->send( $_data);
                        $sendCount++;
                    }
                }
            }
        }
        return $sendCount;
    }

 	/**
     * Check errors when current process exited.
     *
     * @return void
     */
    public static function checkErrors()
    {
        if (self::STATUS_SHUTDOWN != self::$_status) {
            $error_msg = "WORKER EXIT UNEXPECTED ";
            $errors    = error_get_last();
            if ($errors && ($errors['type'] === E_ERROR ||
                    $errors['type'] === E_PARSE ||
                    $errors['type'] === E_CORE_ERROR ||
                    $errors['type'] === E_COMPILE_ERROR ||
                    $errors['type'] === E_RECOVERABLE_ERROR)
            ) {
            	//getErrorType
                $error_msg .= $errors['type']. " {$errors['message']} in {$errors['file']} on line {$errors['line']}";
            }
            self::log($error_msg);
            self::log_notify(false,$error_msg,self::$logFile);
        }
    }

 	public static function log($msg)
    {
        $msg = $msg . "\n";
        if (!self::$daemonize) {
            echo $msg;
        }
        file_put_contents(self::$logFile, date('Y-m-d H:i:s') . ' ' . 'pid:'. posix_getpid() . ' ' . $msg, FILE_APPEND | LOCK_EX);
    }

    public  static function log_notify( $_title, $_errMsg, $_file = false)
    {
        $mail = new \OM\Lib\Mailer;

        $mail->isSMTP();
        $mail->Host = 'smtp.ym.163.com';
        $mail->SMTPAuth = true;
        $mail->Username = 'notify@bizz2.info';
        $mail->Password = 'D_hh168168';
        $mail->SMTPSecure = 'tls';
        $mail->Port = 25;

        $mail->setFrom('notify@bizz2.info', '维护通知邮件');
        $mail->addAddress('1295918243@qq.com', 'ZhangYe');
        //$mail->addCC('cc@example.com');
        //$mail->addBCC('bcc@example.com');

        if( $_file ){
            if( is_string( $_file ) && file_exists( $_file )){
                $mail->addAttachment( $_file );
            }elseif( is_array( $_file) ){
                foreach( $_file as $vf ){
                    if( file_exists( $vf ) )
                    $mail->addAttachment( $vf );
                }
            }
        }
        $mail->isHTML( false );
        if( !$_title )
            $_title = '代理程序意外退出——'.date('Y-m-d H:i:s');
        $mail->Subject = $_title;
        $mail->Body    = $_errMsg.self::writeStatisticsToStatusFile();

        return $mail->send();
    }

    /**
     * Install signal handler.
     *
     * @return void
     */
    protected static function installSignal()
    {
        // stop
        pcntl_signal(SIGINT, array('\OM\OMadapter', 'signalHandler'), false);
        // reload
        pcntl_signal(SIGUSR1, array('\OM\OMadapter', 'signalHandler'), false);
        // status
        pcntl_signal(SIGUSR2, array('\OM\OMadapter', 'signalHandler'), false);
        // ignore
        pcntl_signal(SIGPIPE, SIG_IGN, false);
    }

    /**
     * Reinstall signal handler.
     *
     * @return void
     */
    protected static function reinstallSignal()
    {
        // uninstall stop signal handler
        pcntl_signal(SIGINT, SIG_IGN, false);
        // uninstall reload signal handler
        pcntl_signal(SIGUSR1, SIG_IGN, false);
        // uninstall  status signal handler
        pcntl_signal(SIGUSR2, SIG_IGN, false);
        // reinstall stop signal handler
        self::$globalEvent->add(SIGINT, EventInterface::EV_SIGNAL, array('\OM\OMadapter', 'signalHandler'));
        // reinstall  reload signal handler
        self::$globalEvent->add(SIGUSR1, EventInterface::EV_SIGNAL, array('\OM\OMadapter', 'signalHandler'));
        // reinstall  status signal handler
        self::$globalEvent->add(SIGUSR2, EventInterface::EV_SIGNAL, array('\OM\OMadapter', 'signalHandler'));
    }

    /**
     * Signal hander.
     *
     * @param int $signal
     */
    public static function signalHandler($signal)
    {
        switch ($signal) {
            // Stop.
            case SIGINT:
                self::stopAll();
                break;
            // Reload.
            case SIGUSR1:
                self::$_pidsToRestart = self::getAllWorkerPids();
                self::reload();
                break;
            // Show status.
            case SIGUSR2:
                self::writeStatisticsToStatusFile();
                break;
        }
    }

    /**
     * Execute reload.
     *
     * @return void
     */
    protected static function reload()
    {
    	//
    }

    /**
     * Exit current process.
     *
     * @return void
     */
    protected static function exitAndClearAll()
    {
        @unlink(self::$pidFile);
        self::log("OMadapter[" . basename(self::$_startFile) . "] has been stopped");
        exit(0);
    }

    /**
     * 终止所有进程
     *
     * @return void
     */
    public static function stopAll()
    {
 		self::$_status = self::STATUS_SHUTDOWN;
        // For master process.
        if (self::$_masterPid === posix_getpid()) {
            self::log("OM[" . basename(self::$_startFile) . "] Stopping ...");
            $worker_pid_array = self::getAllWorkerPids();
            // Send stop signal to all child processes.
            foreach ($worker_pid_array as $worker_pid) {
                posix_kill($worker_pid, SIGINT);
                Timer::add(self::KILL_WORKER_TIMER_TIME, 'posix_kill', array($worker_pid, SIGKILL), false);
            }
        } // For child processes.
        else {
            // Execute exit.
            foreach (self::$_workers as $worker) {
                $worker->stop();
            }
            exit(0);
        }
    }

    /**
     * 讲OM适配器统计数数据写到磁盘文件中
     *
     * @return void
     */
    protected static function writeStatisticsToStatusFile()
    {
        // For master process.
        if (self::$_masterPid === posix_getpid()) {
            $loadavg = sys_getloadavg();
            file_put_contents(self::$_statisticsFile,
                "---------------------------------------GLOBAL STATUS--------------------------------------------\n");
            file_put_contents(self::$_statisticsFile,
                'version:' . OMadapter::VERSION . "          PHP version:" . PHP_VERSION ."        Memory:".ini_get('memory_limit')."\n", FILE_APPEND);
            file_put_contents(self::$_statisticsFile, 'start time:' . date('Y-m-d H:i:s',
                    self::$_globalStatistics['start_timestamp']) . '   run ' . floor((time() - self::$_globalStatistics['start_timestamp']) / (24 * 60 * 60)) . ' days ' . floor(((time() - self::$_globalStatistics['start_timestamp']) % (24 * 60 * 60)) / (60 * 60)) . " hours   \n",
                FILE_APPEND);
            $load_str = 'load average: ' . implode(", ", $loadavg);
            file_put_contents(self::$_statisticsFile,
                str_pad($load_str, 33) . "\n", FILE_APPEND);
            file_put_contents(self::$_statisticsFile,
                count(self::$_pidMap) . ' workers       ' . count(self::getAllWorkerPids()) . "  processes\n",
                FILE_APPEND);
            file_put_contents(self::$_statisticsFile,
                str_pad('worker_name', self::$_maxWorkerNameLength) . " exit_status     exit_count\n", FILE_APPEND);
            foreach (self::$_pidMap as $worker_id => $worker_pid_array) {
                $worker = self::$_workers[$worker_id];
                if (isset(self::$_globalStatistics['worker_exit_info'][$worker_id])) {
                    foreach (self::$_globalStatistics['worker_exit_info'][$worker_id] as $worker_exit_status => $worker_exit_count) {
                        file_put_contents(self::$_statisticsFile,
                            str_pad($worker->name, self::$_maxWorkerNameLength) . " " . str_pad($worker_exit_status,
                                16) . " $worker_exit_count\n", FILE_APPEND);
                    }
                } else {
                    file_put_contents(self::$_statisticsFile,
                        str_pad($worker->name, self::$_maxWorkerNameLength) . " " . str_pad(0, 16) . " 0\n",
                        FILE_APPEND);
                }
            }
            file_put_contents(self::$_statisticsFile,
                "---------------------------------------OM STATUS--------------------------------------------\n",FILE_APPEND);
            if( count(self::$om_ipMap ) ){
                foreach( self::$om_ipMap as $kom=>$vom ){
                    if( self::$om_devices[$kom]->states == Om::OM_RUNNING){
                        file_put_contents(self::$_statisticsFile,
                            "OM ID:".str_pad($kom,5)."\t addr:".str_pad($vom,16) . str_pad(' RUNNIG', 10). "\n",FILE_APPEND);
                    }else{
                        file_put_contents(self::$_statisticsFile,
                            "OM ID:".str_pad($kom,5)."\t addr:".str_pad($vom,16) . str_pad(' OFFLINE', 10). "\n",FILE_APPEND);
                    }
                }
            }
            else{
                file_put_contents(self::$_statisticsFile,"No OM Devices\n",FILE_APPEND);
            }
            file_put_contents(self::$_statisticsFile,
                "---------------------------------------PROCESS STATUS-------------------------------------------\n",
                FILE_APPEND);
            file_put_contents(self::$_statisticsFile,
                "pid\tmemory  " . str_pad('listening', self::$_maxSocketNameLength) . " " . str_pad('worker_name',
                    self::$_maxWorkerNameLength) . " connections " . str_pad('total_request',
                    13) . " " . str_pad('send_fail', 9) . " " . str_pad('throw_exception', 15) . "\n", FILE_APPEND);

            chmod(self::$_statisticsFile, 0722);

            foreach (self::getAllWorkerPids() as $worker_pid) {
                posix_kill($worker_pid, SIGUSR2);
            }
            return file_get_contents( self::$_statisticsFile );
        }

        // For child processes.
        /** @var Worker $worker */
        $worker           = current(self::$_workers);
        $wrker_status_str = posix_getpid() . "\t" . str_pad(round(memory_get_usage(true) / (1024 * 1024), 2) . "M",
                7) . " " . str_pad($worker->getSocketName(),
                self::$_maxSocketNameLength) . " " . str_pad(($worker->name === $worker->getSocketName() ? 'none' : $worker->name),
                self::$_maxWorkerNameLength) . " ";
        $wrker_status_str .= str_pad(ConnectionInterface::$statistics['connection_count'],
                11) . " " . str_pad(ConnectionInterface::$statistics['total_request'],
                14) . " " . str_pad(ConnectionInterface::$statistics['send_fail'],
                9) . " " . str_pad(ConnectionInterface::$statistics['throw_exception'], 15) . "\n";
        file_put_contents(self::$_statisticsFile, $wrker_status_str, FILE_APPEND);
        return file_get_contents( self::$_statisticsFile );
    }

 	/**
 	 *  Run OM as a daemon.
 	 *  How to daemonize a process,please reffer Advanced Programming in the UNIX Environment - chapter 13
 	 *
 	 * @return [type] [description]
 	 */
 	protected static function daemonize()
 	{
 		if (!self::$daemonize) {
            return;
        }
 		umask(0);
    	$pid = pcntl_fork();
    	if (-1 === $pid) {
        throw new Exception('fork fail');
    	} elseif ($pid > 0) {
        exit(0);
    	}

    	if (-1 === posix_setsid()) {
        throw new Exception("setsid fail");
    	}
    	// Fork again avoid SVR4 system regain the control of terminal.
    	$pid = pcntl_fork();
    	if (-1 === $pid) {
        	throw new Exception("fork fail");
    	} elseif (0 !== $pid) {
        exit(0);
    	}
    	self::log( self::LOG_NOTICE."Run OM In daemon mode");
 	}

 	 protected static function resetStd()
    {
        if (!self::$daemonize) {
            return;
        }
        global $STDOUT, $STDERR;
        $handle = fopen(self::$stdoutFile, "a");
        if ($handle) {
            unset($handle);
            @fclose(STDOUT);
            @fclose(STDERR);
            $STDOUT = fopen(self::$stdoutFile, "a");
            $STDERR = fopen(self::$stdoutFile, "a");
        } else {
            throw new Exception('can not open stdoutFile ' . self::$stdoutFile);
        }
    }

 	protected static function saveMasterPid()
    {
        self::$_masterPid = posix_getpid();
        if (false === @file_put_contents(self::$pidFile, self::$_masterPid)) {
            throw new Exception('can not save pid to ' . self::$pidFile);
        }
    }

   	protected static function getCurrentUser()
    {
        $user_info = posix_getpwuid(posix_getuid());
        return $user_info['name'];
    }

    /**
     * Get all pids of worker processes.
     *
     * @return array
     */
    protected static function getAllWorkerPids()
    {
        $pid_array = array();
        foreach (self::$_pidMap as $worker_pid_array) {
            foreach ($worker_pid_array as $worker_pid) {
                $pid_array[$worker_pid] = $worker_pid;
            }
        }
        return $pid_array;
    }

    protected static function setProcessTitle($title)
    {
        // >=php 5.5
        if (function_exists('cli_set_process_title')) {
            @cli_set_process_title($title);
        } // Need proctitle when php<=5.5 .
        elseif (extension_loaded('proctitle') && function_exists('setproctitle')) {
            @setproctitle($title);
        }
    }

    public function getSocketName()
    {
        return $this->_socketName ?  $this->_socketName : 'none';
    }

   	//run instance
    public function run()
    {
    	self::$_status = self::STATUS_RUNNING;
    	register_shutdown_function(array("\\OM\\OMadapter", 'checkErrors'));
    	Loader::setRootPath($this->_autoloadRootPath);

    	//create a global event loop
    	if (!self::$globalEvent) {
            self::$globalEvent = new \OM\Events\Event ;
            // Register a listener to be notified when server socket is ready to read.
            if ($this->_socketName) {
                self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ,array($this, 'acceptConnection'));
            }
        }

        // Reinstall signal.
        self::reinstallSignal();

        // Init Timer.
        Timer::init(self::$globalEvent);

        // Try to emit onStart callback.
        if ($this->onStart) {
            try {
                call_user_func($this->onStart, $this);
            } catch (\Exception $e) {
                self::log($e);
                exit(250);
            } catch (\Error $e) {
                self::log($e);
                exit(250);
            }
        }

        // Main loop.
        self::$globalEvent->loop();
    }


    public function listen()
    {
        if (!$this->_socketName || $this->_mainSocket) {
            return;
        }

        // Autoload.
        Loader::setRootPath($this->_autoloadRootPath);

        // Flag.
        $flags  = STREAM_SERVER_BIND | STREAM_SERVER_LISTEN;
        $errno  = 0;
        $errmsg = '';
        // SO_REUSEPORT.
        if ($this->reusePort) {
            stream_context_set_option($this->_context, 'socket', 'so_reuseport', 1);
        }

        // Create an Internet or Unix domain server socket.
        // $this->_socketName;
        $this->_mainSocket = stream_socket_server( $this->_socketName , $errno, $errmsg, $flags, $this->_context);
        if (!$this->_mainSocket) {
            throw new \Exception($errmsg);
        }

        // Try to open keepalive for tcp and disable Nagle algorithm.
        if (function_exists('socket_import_stream')  ) {
            $socket = socket_import_stream($this->_mainSocket);
            @socket_set_option($socket, SOL_SOCKET, SO_KEEPALIVE, 1);
            @socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1);
        }

        // Non blocking.
        stream_set_blocking($this->_mainSocket, 0);

        // Register a listener to be notified when server socket is ready to read.
        if (self::$globalEvent) {
            self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptConnection'));
        }
    }

    public function stop()
    {
        // Try to emit onStop callback.
        if ($this->onStop ) {
            try {
                call_user_func($this->onStop, $this);
            } catch (\Exception $e) {
                self::log($e);
                exit(250);
            } catch (\Error $e) {
                self::log($e);
                exit(250);
            }
        }
        // Remove listener for server socket.
        self::$globalEvent->del($this->_mainSocket, EventInterface::EV_READ);
        @fclose($this->_mainSocket);
    }

	/**
	* Accept a connection.
	*
	* @param resource $socket
	* @return void
	*/
    public function acceptConnection($socket)
    {
        // Accept a connection on server socket.
        $new_socket = @stream_socket_accept($socket, 0, $remote_address);
        // Thundering herd.
        if (!$new_socket) {
            return;
        }
        // TcpConnection.
        $connection                         = new TcpConnection($new_socket, $remote_address);
        $this->m_connections[$connection->id] = $connection;
        $connection->worker                 = $this;
        $connection->protocol               = "\\OM\Lib\\Protocols\\Http";
        $connection->onMessage              = $this->onMessage;
        $connection->onClose                = $this->onClose;
        $connection->onError                = $this->onError;
        $connection->onBufferDrain          = $this->onBufferDrain;
        $connection->onBufferFull           = $this->onBufferFull;

        //Websocket 事件绑定
        $connection->onWebSocketConnect    = $this->onWebSocketConnect;

        // Try to emit onConnect callback.
        if ($this->onConnect) {
            try {
                call_user_func($this->onConnect, $connection);
            } catch (\Exception $e) {
                self::log($e);
                exit(250);
            } catch (\Error $e) {
                self::log($e);
                exit(250);
            }
        }
    }
 }
?>